package com.gupaoedu.kafka.kafkapractice;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @Author: JLL
 * @Date: 2019/10/30 16:52
 */
public class GpKafkaProducer extends Thread{

    KafkaProducer<Integer,String> producer;
    String topic;

    public GpKafkaProducer(String topic) {

        Properties properties = new Properties();
        //为了解决你不知道配哪些key，它把key都给你定义好了，你直接给value就行。
        //连接字符串
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.150.130:9092");
        //client_id
        properties.put(ProducerConfig.CLIENT_ID_CONFIG,"gp-producer");

        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,"");
        properties.put(ProducerConfig.LINGER_MS_CONFIG,"");


        //key的序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        //value的序列化 、 这些都是告诉producer用什么序列化的方式去传输
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


        //要么工厂，要么实例。
        producer = new KafkaProducer<Integer, String>(properties);
        this.topic = topic;
    }

    @Override
    public void run() {
        int num= 0;
        String msg = "gp kafka practice msg" + num ;
        //模拟发多条消息
        while (num<20){

        }
        //topic是发消息给哪个地方？？？  应该是分类吧？？
        //get 会拿到发送的结果
        try {
            //这个发送方式是同步。get是阻塞方法  但是消息中间件本应该要 异步去处理。  DUBBO也是异步去处理的
            //如果是异步发送，那肯定需要一个回调通知
            /*RecordMetadata recordMetadata =
                    producer.send(new ProducerRecord<>(topic,msg)).get();*/
            /*producer.send(new ProducerRecord<>(topic, msg), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {

                }
            });*/
            //如果是异步发送，那肯定需要一个回调通知
            producer.send(new ProducerRecord<>(topic, msg), (RecordMetadata recordMetadata, Exception e) -> {
                System.out.println(recordMetadata.offset() + " - " + recordMetadata.partition() + " - " + recordMetadata.topic());
            });
            TimeUnit.SECONDS.sleep(2);
            ++num;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    public static void main(String[] args) {
        new GpKafkaProducer("test").start();
    }
}
